package com.happy.xiao.common.rabbitmq;

import com.happy.xiao.common.rabbitmq.bean.MessageReliabilityConfirm;
import com.happy.xiao.common.rabbitmq.db.service.IMessageFailLogService;
import com.happy.xiao.common.rabbitmq.service.IReliabilityMessage;
import com.happy.xiao.common.rabbitmq.util.MessageUtil;
import com.happy.xiao.common.redis.IRedisService;
import com.happy.xiao.common.utils.Util;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.util.StringUtils;

import java.io.IOException;

/**
 * 这里是所有消费者需要继承的listener
 *
 */
public class BaseConsumerListener {

    @Autowired
    private Environment env;

    @Autowired
    IReliabilityMessage iReliabilityMessage;


    @Autowired
    IRedisService iRedisService;

    @Autowired
    IMessageFailLogService iMessageFailLogService;


    // 默认退回队列次数
    private static final int DEFAULT_RECOVER_COUNT = 7;

    /**
     * ####处理成功信息，手动确认以及清楚Redis缓存####
     * ####这个方法与handleErrorMessage 方法联用####123
     *
     * @param channel     渠道对象
     * @param message     消息对象
     * @param redisPrefix Redis前缀
     * @throws IOException
     */
    public void handleSuccessMessage(Channel channel,
                                     Message message,
                                     String redisPrefix)
            throws IOException {

        if (channel == null || message == null) return;

        MessageProperties messageProperties = message.getMessageProperties();
        long deliveryTag = messageProperties.getDeliveryTag();      // 消息的标识
        String messageId = messageProperties.getMessageId();        // 消息的ID

        String receivedExchange = messageProperties.getReceivedExchange();
        String receivedRoutingKey = messageProperties.getReceivedRoutingKey();
        String expiration = messageProperties.getExpiration();    // 过期时间
        int delayTime = StringUtils.isEmpty(expiration) ? 0 : Integer.parseInt(expiration);

        String messageContent = MessageUtil.getMessageContent(message);

        // 清空这个缓存, 这里注释掉。让这个redis值自动死亡，减少一次操作redis行为，
//        if (!StringUtils.isEmpty(messageId)) iRedisService.delete(redisPrefix + messageId);

        // 手动确认
        channel.basicAck(deliveryTag, true);

        // 以下是执行，发送 【消费确认】 的消息， 发送消费确认队列消息这个行为，
        // 要把【自身】的发送消费确认行为给排除掉，这个【很重要】，不然会【循环发送】。
        // 这里是消费确认的队列
        String confirmExchange = env.getProperty("check.confirm.exchange.name");
        String confirmRoutingKey = env.getProperty("check.confirm.routing.key.name");

        String checkExchange = env.getProperty("check.dead.exchange.name");
        String checkRoutingKey = env.getProperty("check.dead.routing.key.name");

        if (receivedExchange.equals(confirmExchange)) return;
        if (receivedRoutingKey.equals(confirmRoutingKey)) return;

        if (receivedExchange.equals(checkExchange)) return;
        if (receivedRoutingKey.equals(checkRoutingKey)) return;

        try {
            MessageReliabilityConfirm confirm = new MessageReliabilityConfirm(messageId, receivedExchange,
                    receivedRoutingKey, messageContent, delayTime);

            String confirmJson = Util.ObjectToJson(confirm);
            if (StringUtils.isEmpty(confirmJson)) return;
            // 这里发送消息确认是否消费成功
            iReliabilityMessage.sendReliabilityConfirmMessage(confirmJson);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }







    /**
     * RabbitMq消息异常统一处理
     *
     * @param channel      渠道对象
     * @param message      消息对象
     * @param redisPrefix  redis的前缀
     * @param recoverCount 重回消息队列次数
     * @throws IOException
     */
    public void handleErrorMessage(Channel channel,
                                   Message message,
                                   String redisPrefix,
                                   int recoverCount)
            throws IOException {
        if (channel == null || message == null) return;

        // 这里是消息的数据
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String messageId = message.getMessageProperties().getMessageId();
        String receivedRoutingKey = message.getMessageProperties().getReceivedRoutingKey();
        String receivedExchange = message.getMessageProperties().getReceivedExchange();
        String expiration = message.getMessageProperties().getExpiration();  // 过期时间

        String msgStr = MessageUtil.getMessageContent(message);

        if (recoverCount == 0) recoverCount = 3;

        // TODO: 这里要用缓存控制一下，重入队列
        // 先判断redis中，是否存在着个异常消息次数，如果不存在，则添加到redis
        // 判断redis中的次数，如果超过一定的次数，如10次，则把这个消息丢弃

        if (StringUtils.isEmpty(messageId)) {
            channel.basicReject(deliveryTag, false);
            return;
        }
        String redisKey = redisPrefix + messageId;
        Object msgIDObj = iRedisService.getKeyValue(redisKey);

        int msgIDCount = 1;
        if (msgIDObj != null) msgIDCount = (int) msgIDObj;

        // 如果这个
        if (msgIDCount < recoverCount) {
            iRedisService.increment(redisKey, 1,  7 * 86400);
            // 回滚到队列中
            channel.basicNack(deliveryTag, false, true);
        } else {
            // 把这个消息丢弃了，因为上面已经重发了
            channel.basicReject(deliveryTag, false);
            // 这里是把丢失的数据。保存到发送失败的表中
            iMessageFailLogService.createFailMessageInDB(messageId, receivedExchange, receivedRoutingKey, msgStr, expiration, 1);
            // 清空这个缓存
            iRedisService.delete(redisKey);
        }
    }

    /**
     * RabbitMq消息异常统一处理
     *
     * @param channel     渠道对象
     * @param message     消息传送的ID
     * @param redisPrefix redis的前缀
     * @throws IOException
     */
    public void handleErrorMessage(Channel channel,
                                   Message message,
                                   String redisPrefix)
            throws IOException {
        handleErrorMessage(channel, message, redisPrefix, DEFAULT_RECOVER_COUNT);
    }

}
